#include "clocksync.h"

long cycles[USERS], sent[USERS];
long double tm_virt=0.0, tm_init=0.0, tm_clock = 0.0;		//Virtual Clock being used
clock_t st_time, curr_time,  t_virt, t_init;			//Virtual clocks in clock_t format. This hasnt been used finally for tests and hence can be gotten rid of. I did not want to end up tampering with things to left it unchanged
FILE *f[USERS];						//A separate file created for logging information between every two nodes.
char filenam[USERS][20], filenam_t[20];			//File names being stored in here.
long double t_diff, t_max[USERS], t_min[USERS], t_avg[USERS];	//Same as variables below but for type clock_t and so can be gotten rid of.
/*
tm_diff has the round trip message exchange time, ie time taken by the messages for travel only
tm_min stores the minimum t_diff between every two nodes,
similarly avg and max for t_min and t_max respectively. 
tm_maxcycle stores the round trip time for the exchange of message M and ACK
tm_minrecvr stores minimum time at the receiver's end
*/
long double tm_diff, tm_max[USERS], tm_min[USERS], tm_avg[USERS], tm_maxcycle[USERS], tm_minrecv[USERS];

pthread_mutex_t mutex10 = PTHREAD_MUTEX_INITIALIZER;

//Input the node of the talker when calling this function
void start_clocksync(int argc, char *argv[])
{
  int ret3, ret4, ret5, ret6, ret7, ret8, i;
  pthread_t appthsend, appthrecv, appsendrecv, thread_time, th_clklog, th_listen;
  if(argc!=2)
    {
      fprintf(stderr, "usage: talker\'s_node_number\n");
      exit(1);
    }
  strcpy(filenam_t, argv[1]);
  strcat(filenam_t, "syncs");

  for(i=0;i<USERS;i++)
    {
      strcpy(filenam[i], filenam_t);	//Binding filenames
      strcat(filenam[i], node[i]);

      if((f[i]=fopen(filenam[i], "w"))==NULL)
	{
	  printf("Can't open file to write\n");
	  exit(1);
	}
	fprintf(f[i], "remote node number, iteration, real local clock, virtual local clock, min(t_rt), avg(t_rt), max(t_rt), number of msgs sent, number of msgs received\n");
    }
  if((ret5=pthread_create( &appsendrecv, NULL, &main_sendrecv, (void *)argv[1])))
   {
      	printf("Thread creation failed: %d\n", ret5);
   }
  t_virt = t_init = st_time = clock();
  tm_virt = tm_init = tm_clock = 0.0;
  if((ret6=pthread_create( &thread_time, NULL, &time_virtual, NULL)))
   {
      	printf("Thread creation failed: %d\n", ret6);
   }
  sleep(1);					//Essential for all initialisations to complete
  fopen_listen(argv[1]);			//Opening file to listen
  if(( ret8=pthread_create( &th_listen, NULL, &applisten, NULL) ))
   {
        printf("Thread creation failed: %d\n", ret7);
   }
  if((ret3=pthread_create( &appthsend, NULL, &appsendclk, NULL)))
   {
      	printf("Thread creation failed: %d\n", ret3);
   }
  if((ret4=pthread_create( &appthrecv, NULL, &apprecvclk, NULL)))
   {
        printf("Thread creation failed: %d\n", ret4);
   }
  if( (ret7=pthread_create( &th_clklog, NULL, &app_clklog, (void *)argv[1])) )
   {
        printf("Thread creation failed: %d\n", ret7);
   }

   pthread_join(appthsend, NULL);
   pthread_join(appthrecv, NULL);
   close(sockfd);
   return;
}

void *app_clklog(void *nodeno)		//Function which interacts with logging data
{
  int sockfd_clklog, numbytes;
  char my_add[16], *nodeno_ch;
  unsigned add_len;
  unsigned long j=0;
  msg_clklog *logmsg;
  char *msg_ch;
  clock_t start_time, current_time;
  struct sockaddr_in my_addr, their_addr;
  nodeno_ch = (char *)nodeno;		//Node no in char form
  logmsg = (msg_clklog *)malloc(sizeof(msg_clklog));
  msg_ch = (char *)malloc(sizeof(msg_clklog) + sizeof(char));	//Exchanging message in char form. Can be gotten rid of. Did not because did not want to take a chance with some other issue coming up
  if((sockfd_clklog = socket(AF_INET, SOCK_DGRAM, 0))==-1)
    {
      perror("socket");
      exit(1);
    }
  strcpy(my_add, ipclass_w);
  strcat(my_add, nodeno_ch);

  my_addr.sin_family = AF_INET;
  my_addr.sin_port = htons(PORT_LOG);
  my_addr.sin_addr.s_addr = inet_addr(my_add);
  memset(&(my_addr.sin_zero), '\0', 8);

  if(bind(sockfd_clklog, (struct sockaddr *)&my_addr, sizeof(struct sockaddr))==-1)
    {
      perror("bind");
      exit(1);
    }

  while(1)
    {
      add_len = sizeof(struct sockaddr);
      if((numbytes=recvfrom(sockfd_clklog, msg_ch, sizeof(msg_clklog) + sizeof(char), 0, (struct sockaddr *) &their_addr, &add_len)) == -1)
	{
	  perror("recvfrom");
	  continue;
	}
      else if(numbytes >0)
	{
	  logmsg = (msg_clklog *)msg_ch;
	  printf("Received logmsg: %lu, %llf\n", j, logmsg->virt_time);
	  if(j==0)		//If first message is received, reset clock to 0
	    {
	      st_time = clock();
	      pthread_mutex_lock( &mutex10 );
	      tm_clock = tm_init = tm_virt = 0.0;
	      pthread_mutex_unlock( &mutex10 );
	    }
	  current_time = clock(); //t_virt + clock() - t_init;
//	  logmsg->curr_time = current_time - st_time;
	  logmsg->virt_time = tm_clock + tm_virt - tm_init;		//Returning virtual time(last updated virtual time + time gone by from last update)
	  if(sendto(sockfd_clklog, logmsg, sizeof(msg_clklog), 0, (struct sockaddr *) &their_addr, sizeof(struct sockaddr))==-1)
	    {
	      perror("sendto");
	      exit(1);
	    }
	  else
	    printf("Sent logmsg: %lu AT: %llf\n", j, logmsg->virt_time);
	  j++;
	}
    }
  free(logmsg);
}

//User Defined Function to keep time. Sleeps for a bit and then updates by the same amount
void *time_virtual()
{
  tm_virt = tm_init = tm_clock = 0.0;
  st_time = clock();
  while(1)
  {
    usleep(2);
    tm_clock = tm_clock + .000020;
    curr_time = clock();
  }
}
/*
int musleep(unsigned long microsec)		//This function could be used in place of usleep
{
    struct timespec req={0};
    time_t sec=(int)(microsec/1000000L);
    microsec=microsec-(sec*1000000L);
    req.tv_sec=sec;
    req.tv_nsec=microsec*1000;
    while(nanosleep(&req,&req)==-1)
      continue;
    return 1;
}
*/
//Application before sending to queue
void *appsendclk()
{
  int i;
  unsigned long j;
  msg_clksync *msg;
  struct sockaddr_in dest_add;
  for(i=0;i<USERS;i++)
    sent[i]=0;
  for(j=0;1;j++)	//Sending messages forever at regular intervals
    {
      sleep(D_T);
      for(i=0;i<USERS;i++)	//Sending to all nodes
	{
	  if(i==hno)		//If wanting to send to self 
	    continue;
	  msg = (msg_clksync *)malloc(sizeof(msg_clksync));
	  msg->type = SEND_SYNC1;
	  msg->t0 = t_virt + clock() - t_init;
	  msg->tm0 = tm_virt + tm_clock - tm_init;	//Time, t_send
	  msg->machine = i;
	  msg->iteratn = j;
	  dest_add = addr[i];
	  queue_send(msg, sizeof(msg_clksync), dest_add);	//Calling queue_send()
	  sent[i]++;
	  //printf("Sent %d, %lu\n", i, j);
	  free(msg);
	}
    }
  return (NULL);
}

//Application for receiving and working with the receives
void *apprecvclk()
{
  msg_clksync* msg;
  pthread_t print_time;
  clock_t tstrt, tsnd, trcv, tend;
  long double tmstrt, tmsnd, tmrcv, tmend;
  int i, size_msg, ret7;
  char *msg_ch;
  unsigned long j;
  msg_s *mw;
  t_print *t_ime;
  t_ime = (t_print *)malloc(sizeof(t_print));
  struct sockaddr_in their_addr;

  for(i=0;i<USERS;i++)		//Initializations
    {
      t_max[i]=0.0;
      t_min[i]=999999.0;
      t_avg[i]=0.0;
      tm_max[i]=0.0;
      tm_min[i]=999999.0;
      tm_avg[i]=0.0;
      cycles[i]=0;
      tm_maxcycle[i] = 0.0;
      tm_minrecv[i] =99999.0;
    }
  while(1)
    {
      msg_ch = (char *)malloc(sizeof(msg_s) + sizeof(char));			//Message exchanged in type char. Here it is essential unlike above, due to the use of queue_recv.
      if((queue_recv(msg_ch, RECV_SYNC1, &size_msg, &their_addr))>0)            //If queue is non-empty
	{
	  tmend = tm_virt + tm_clock - tm_init;
	  mw = (msg_s *)msg_ch;
	  tsnd = mw->t2;
	  tmsnd = mw->tm2;
	  trcv = mw->msw.t1;
	  tmrcv = mw->msw.tm1;
	  tstrt = mw->msw.m.t0;
	  tmstrt = mw->msw.m.tm0;
	  i = mw->msw.m.machine;
	  j = mw->msw.m.iteratn;
	  mw = NULL;
	  t_diff = ((tend - tstrt - tsnd + trcv + 0.00L)/(2.0 * CLOCKS_PER_SEC));
	  tm_diff = ((tmend - tmstrt - tmsnd + tmrcv)/2.0);	//One side message transfer time
	  if(tmend - tmstrt > tm_maxcycle[i])
	      tm_maxcycle[i] = tmend - tmstrt;			//Max cycle time, time taken for M and ACK to complete
	  if( (tmsnd - tmrcv) < tm_minrecv[i])			//Time message was at the receiver for
		tm_minrecv[i] = (tmsnd - tmrcv);
/*
Calculating avg, min and max one side time
*/
	 if(tm_diff>tm_max[i])
	    {
		tm_max[i]=tm_diff;
		t_max[i]=t_diff;
	    }
	  if(tm_diff<tm_min[i])
	    {
		t_min[i]=t_diff;
		tm_min[i] = tm_diff;
	    }
	  t_avg[i]=(t_avg[i]*cycles[i] + t_diff)/(cycles[i]+1);
	  tm_avg[i]=(tm_avg[i]*cycles[i] + tm_diff)/(cycles[i]+1);

	  cycles[i]++;		//Cycles of message exchange completed

          printf("\nMachine:%d\tIteration:%lu\n, Max: %llf, MaxCycle: %llf, MinTimeRecvr: %llf [Times denoted in msecs], Packets received: %ld out of %ld\n", i, j, tm_max[i], tm_maxcycle[i], tm_minrecv[i], cycles[i], sent[i]);

	  if((tmsnd + tm_avg[i] ) > tmend)      //If the clock of the client is faster
	  {
		tm_virt = tmsnd + tm_avg[i] + tm_clock - tm_init;
		t_virt = tsnd + (clock_t)(t_avg[i]*CLOCKS_PER_SEC) + clock() - t_init;
	  }
	  else                                //If clock of server is faster
	  {
	    t_virt = t_virt + clock() - t_init;
	    tm_virt = tm_virt + tm_clock - tm_init;
	  }
	  t_init = clock();
	  tm_init = tm_clock;
	  msg = (msg_clksync*)malloc(sizeof(msg_clksync));	//Learn message created
	  msg->type = LEARN_SYNC;
	  msg->t0 = t_virt + (clock_t)(t_avg[i] * CLOCKS_PER_SEC);
	  msg->tm0 = tm_virt + tm_avg[i];			//Time sent to listener for synchronizing
	  msg->iteratn = j;                   //Value of iteratn is a flag to signify that this message is to synchronize the clocks
	  msg->machine = i;
	  queue_send(msg, sizeof(msg_clksync), addr[i]);
	  free(msg);
	  t_ime->avg_time = t_avg[i];
	  t_ime->avg_tm = tm_avg[i];
	  t_ime->machine = i;
	  t_ime->iteratn = j;
	  if((ret7=pthread_create( &print_time, NULL, &time_print, (void *)(t_ime) )))
	    {
	      printf("Thread creation failed: %d\n", ret7);
	    }
	  else
	    pthread_join(print_time, NULL);
	}
      else
        usleep(1000);
      free(msg_ch);
    }
  return (NULL);
}

void *time_print(void *ptr_tmp)			//Prints time synchronously on different nodes
{
  int i;
  t_print time_a, *ptr_time;
  ptr_time = (t_print *)ptr_tmp;
  time_a = *ptr_time;

  if(time_a.avg_time>0)
    {
      sleep((int)(time_a.avg_tm));                    //To print the two messages simultaneously
      usleep((time_a.avg_tm - (int)(time_a.avg_tm) + 0.00L)*1000000L);
    }
  i = time_a.machine;
  t_virt += clock() - t_init;
  t_init = clock();
  tm_virt += tm_clock - tm_init;
  tm_init = tm_clock;
  fprintf(f[i], "%d, %lu, %llf, %llf, %llf, %llf, %llf, %ld, %ld\n", time_a.machine, time_a.iteratn, tm_init , tm_virt, tm_min[i], tm_avg[i], tm_max[i], cycles[i], sent[i]);	//Fprintf could be replaced by printf to see somewhat synchronized printing.
  fflush(f[i]);
  return(NULL);
}
